# coding=utf-8
# Date: Jan 24th, 2021
# Author: geozhang
import json
import uuid

import pika

# rabbitmq 配置信息
from pika.exceptions import NackError, UnroutableError

from rabbitmq.mq_utils import json_to_dict

MQ_CONFIG = {
    "host": "114.55.104.203",
    "port": 5672,
    "virtual_host": "/",
    "user": "guest",
    "password": "guest",

    "request_queue01": "request-python-queue",
    "request_exchange01": "request-python-exchange",
    "request_routing_key01": "request-routing-key",

    "response_queue01": "response-python-queue",
    "response_exchange01": "response-python-exchange",
    "response_routing_key01": "response-routing-key",

    "content_type": "application/json",

}


class MQBase(object):
    """ 消息队列基类 """

    def __init__(self, host, port, virtual_host='/', auto_ack=False, durable=True):
        self.conn = None  # TCP连接
        self.channel = None  # 信道，逻辑连接
        self.properties = None

        self.host = host  # 地址
        self.port = port  # 端口
        self.virtual_host = virtual_host  # 虚拟空间
        # self.exchange = exchange  # 交换机名称
        # self.exchange_type = exchange_type  # 交换机类型
        self.auto_ack = auto_ack  # 是否开启消息自动确认机制，默认False
        self.durable = durable  # 消息是否持久化

    def _open_channel(self):
        """ 建立连接并开辟信道 """

        if not self.conn:
            self.conn = pika.BlockingConnection(
                pika.ConnectionParameters(host=self.host,
                                          port=self.port,
                                          virtual_host=self.virtual_host))  # 建立连接

        if not self.channel:
            self.channel = self.conn.channel()  # 在连接中开辟信道

            if not self.auto_ack:
                self.channel.confirm_delivery()  # 在该信道中开启消息手动确认机制

    def _close_channel(self):
        """ 关闭信道并断开连接 """

        if self.channel and self.channel.is_open:  # 检测信道是否还存活
            self.channel.close()  # 关闭信道

        if self.conn and self.conn.is_open:  # 检测连接是否还存活
            self.conn.close()  # 断开连接


class Provider(MQBase):
    """ 消息队列-生产者 """

    def _prepare_provider(self, req_exchange: str, req_routing_key: str,
                          rsp_queue: str, rsp_exchange: str, rsp_routing_key: str):

        # 创建TCP连接、建立通道
        self._open_channel()

        # 声明request的交换机
        self.channel.exchange_declare(exchange=req_exchange, durable=self.durable)

        # 声明response队列、交换机、绑定关系
        self.channel.queue_declare(queue=rsp_queue, durable=self.durable)

        self.channel.exchange_declare(exchange=rsp_exchange, durable=self.durable)

        # response队列通过路由键绑定到交换机, 之后交换机会将对应路由键的消息转发到该队列
        self.channel.queue_bind(queue=rsp_queue,  # 队列
                                exchange=rsp_exchange,  # 交换机
                                routing_key=rsp_routing_key)  # 绑定键

        # provider监听reply_queue
        self.channel.basic_consume(queue=rsp_queue,
                                   on_message_callback=self.response_handler,
                                   auto_ack=True)

        self.channel.confirm_delivery()  # 开启消息送达确认，provider→broker的确认(注意这里是送达消息队列即可)
        print(" [P] preparation finish.")

    def response_handler(self, channel, method, properties, body_json):

        if self.correlation_id == properties.correlation_id:
            # print(body_json)
            self.response_body = json.loads(body_json)

            for response_unit in self.response_body:
                print(" [P] 收到 response{0}, status: {1}"
                      .format(response_unit["id"], response_unit["status"]))

    def call(self, req_exchange, req_routing_key, response_queue, response_exchange,
             response_routing_key, request_body):

        self.correlation_id = str(uuid.uuid4())
        print(" [P] correlation id: " + self.correlation_id)

        # 生产者准备
        self._prepare_provider(req_exchange,
                               req_routing_key,
                               response_queue,
                               response_exchange,
                               response_routing_key)

        try:
            # 创建请求头
            request_properties = pika.BasicProperties(
                delivery_mode=(2 if self.durable else 0),
                content_type=MQ_CONFIG.get("content_type"),
                reply_to=response_exchange,
                correlation_id=self.correlation_id)

            # 发送消息
            self.channel.basic_publish(
                exchange=req_exchange,  # 指定发送到的交换机
                routing_key=req_routing_key,  # 消息中的路由键
                body=request_body,  # 消息中的有效载荷
                properties=request_properties,
                mandatory=True)  # 该方法用于发送消息, 消息成功送达消息队列时返回True, 否则返回False
            print(' [P] Message publish was confirmed')
        except NackError as e:
            print(" [P] Nack Error")
        except UnroutableError:
            print('Message could not be confirmed')


        # 等待consumer回复
        self.response_body = None
        while self.response_body is None:
            # 阻塞等待消息。连接的任何消息都可以使它脱离阻塞状态（有点像Ajax的事件等待机制）
            self.conn.process_data_events()

        print(" [P] response处理完成。\n")
        # print(" [P] Response: %s" % str(self.response_body))

        return self.response_body


class Consumer01(MQBase):
    """ 消息队列-消费者01 """

    def _declare_and_binding(self, queue_name: str, exchange_name: str, routing_key: str):
        """ 声明队列；创建裸队列、绑定关系 """

        self.channel.queue_declare(queue=queue_name,  # 队列名称
                                   durable=self.durable)  # 队列是都否持久化; 该方法用于声明队列, 声明多次仅会创建一次

        self.channel.exchange_declare(exchange=exchange_name,  # 交换机名称
                                      durable=self.durable)  # 交换机是否持久化; 该方法用于声明交换机, 声明多次仅会创建一次

        self.channel.queue_bind(queue=queue_name,  # 队列
                                exchange=exchange_name,  # 交换机
                                routing_key=routing_key)  # 绑定键, 该方法用于将队列通过绑定键绑定到交换机, 之后交换机会将对应路由键的消息转发到该队列上

    def _subscribe_queue(self, request_exchange: str, request_queue: str, request_routing_key: str):
        """ 订阅队列；裸队列和绑定关系基础上，定制队列的消费方式 """

        # 声明队列、绑定交换机
        self._declare_and_binding(request_queue, request_exchange, request_routing_key)

        self.channel.basic_qos(prefetch_count=1)  # 该方法起到负载均衡的作用, 表明一次只接受prefetch_count条信息, 直到消息确认后再接收新的
        # 该方法用于订阅队列, 并分配消息处理器
        self.channel.basic_consume(on_message_callback=self.request_handler,  # 收到消息后的回调方法, 即消息处理器
                                   queue=request_queue,  # 订阅的队列名称
                                   auto_ack=self.auto_ack)  # 是否向broker自动确认

    def request_handler(self, channel, method, properties, body):
        """ 收到消息后的回调方法, 即消息处理器
            封装于_subscribe_queue()中
        """

        self.handler_func(channel, properties, body)  # 调用真正的消息处理器

        if not self.auto_ack:
            channel.basic_ack(delivery_tag=method.delivery_tag)  # 手动回复broker，本条消息free to delete

    def start(self, request_exchange: str, request_queue: str, request_routing_key: str,
              handler_func):
        """ 开始消费, 以回调方式 """

        self.handler_func = handler_func  # 获取消息的具体处理方法

        self._open_channel()  # 基类：建立TCP连接
        self._subscribe_queue(request_exchange, request_queue, request_routing_key)  # 消费者类：定制队列

        # 开始监听、消费queue中的消息，阻塞式
        # 该方法虽然是由channel调用的, 一个conn下有多个channel, 但一个conn只能调用一次该方法
        self.channel.start_consuming()

    def end(self):
        """ 结束消费 """

        self.channel.stop_consuming()  # 该方法用于停止消费
        self._close_channel()
